package com.atguigu.dga.governance.assessor.calc;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.constant.DgaConstant;
import com.atguigu.dga.governance.assessor.Assessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.governance.bean.Stage;
import com.atguigu.dga.util.HttpUtil;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.math.BigDecimal;
import java.net.URISyntaxException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * ClassName: DataSkewAssessor
 * Package: com.atguigu.dga.governance.assessor.calc
 * Description:
 *
 * @Author:
 * @Create: 2024/3/22 08:46
 * @Version: 1.0
 */
@Component("DATA_SKEW")
public class DataSkewAssessor extends Assessor {
    @Value("${spark.history.api.url}")
    private String sparkHistoryUrl;
    @Override
    public void checkProblem(AssessParam assessParam, GovernanceAssessDetail governanceAssessDetail) throws ParseException, URISyntaxException, IOException, InterruptedException, org.apache.hadoop.hive.ql.parse.ParseException, SemanticException {
        // 从Spark历史服务器找到当前被考评表对应的Job信息
        // 提取超过指标时间参数的阶段Stage，找出最大耗时的Task，计算其余Task平均耗时
        // TODO 1 接口： http://hadoop102:18080/api/v1/applications/{yarnId}
        //  http://hadoop102:18080/api/v1/applications/{yarnId}/{attemptId}/stages
        //  http://hadoop102:18080/api/v1/applications/{yarnId}/{attemptId}/stages/{stageId}

        /**
         * yarnID---->{yarnID:xxx, attempts:[{},{} 有成功又失败，筛选出一个成功的]}------->attemptID------->[{stage},{stage},筛选出成功的]
         * -------> List<StageId>------->遍历每个成功的StageID----->stageID 中{stageID:xxx, tasks: {xx:{}, xx:{}} }
         */

        // TODO 2 http请求工具类封装HttpUtil

        // TODO 3 排除ods
        if(DgaConstant.DW_LEVEL_ODS.equals(assessParam.getTableMetaInfo().getTableMetaInfoExtra().getDwLevel())) {
            return;
        }

        // TODO 4 排除没有任务实例或者yarnID的
        if(assessParam.getTDsTaskInstance() == null || assessParam.getTDsTaskInstance().getAppLink() == null) {
            return;
        }

        // TODO 5 提取指标参数
        String metricParamsJson = assessParam.getGovernanceMetric().getMetricParamsJson();
        JSONObject paramJsonObj = JSON.parseObject(metricParamsJson);
        Integer paramPercent = paramJsonObj.getInteger("percent");
        Integer paramStageDurSeconds = paramJsonObj.getInteger("stage_dur_seconds");

        // TODO 6 提取yarnID
        String appLink = assessParam.getTDsTaskInstance().getAppLink();

        // TODO 7 提取completed:true 的attemptID  一个yarnID对应一个成功的attemptID
        String completedAttemptId = getCompletedAttemptId(appLink);

        // TODO 8 提取所有StageID： 一个attemptID 对应多个StageID 需要complete的
        List<String> completedStageIdList = getCompletedStageIdList(appLink, completedAttemptId);

        // TODO 9 提取每个StageID下所有tasks, 封装自定义对象Stage对象中
        List<Stage> stageList = getStageList(appLink, completedAttemptId, completedStageIdList);

        // TODO 10 判断
        List<Stage> dataSkewStageList = stageList.stream().filter(
                stage -> stage.getRealPercent() > paramPercent && stage.getMaxTaskDuration() > paramStageDurSeconds
        ).collect(Collectors.toList());

        if(dataSkewStageList.size() > 0) {
            governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
            governanceAssessDetail.setAssessProblem("存在数据倾斜");
        }

        governanceAssessDetail.setAssessComment("所有的Stage概况:" + stageList);


    }

    // TODO 9 提取每个StageID下所哟tasks, 封装自定义对象Stage对象中
    private List<Stage> getStageList(String appLink, String completedAttemptId, List<String> completedStageIdList) throws IOException {
        ArrayList<Stage> stageList = new ArrayList<>();

        // 遍历StageID集合
        for (String stageId : completedStageIdList) {
            String url = sparkHistoryUrl  + "/" + appLink + "/" + completedAttemptId + "/stages/" + stageId;
            // 发送请求
            String resultJsonStr = HttpUtil.get(url);
            List<JSONObject> stageJsonArr = JSON.parseArray(resultJsonStr, JSONObject.class); // stage也可能有失败重试
            for (JSONObject stageJsonObj : stageJsonArr) {
                if("COMPLETE".equals(stageJsonObj.getString("status"))) {
                    // 取所有Task
                    JSONObject taskJsonObj = stageJsonObj.getJSONObject("tasks"); // tasks:{"": {}, "": {}}
                    // 取每一个Task（JSONObject 本质是map结构）取tasks对象中所哟的key
                    Set<String> keySet = taskJsonObj.keySet();

                    //
                    Long sumDuration = 0L;
                    Long maxDuration = 0L;
                    Long count = 0L;
                    for (String key : keySet) {
                        JSONObject task = taskJsonObj.getJSONObject(key); // 拿到每一个Task
                        if("SUCCESS".equals(task.getString("status"))) {
                            // 提取当前Task的所需信息 封装到Stage对象中：maxDur avgDur stageID realPercent
                            Long duration = task.getLong("duration");
                            sumDuration += duration;
                            count++;
                            maxDuration = Math.max(maxDuration, duration);
                        }
                    }

                    // 封装Stage
                    Stage stage = new Stage();
                    stage.setStageId(stageId);
                    stage.setMaxTaskDuration(maxDuration);
                    if(count == 1) {
                        stage.setAvgTaskDuration(sumDuration);
                        stage.setRealPercent(0L);
                    }else {
                        stage.setAvgTaskDuration((sumDuration - maxDuration) / (count - 1));
                        stage.setRealPercent((maxDuration - stage.getAvgTaskDuration()) * 100 / stage.getAvgTaskDuration());
                    }

                    stageList.add(stage);

                }
            }
        }

        return stageList;
    }

    // TODO 8 提取所有StageID： 一个attemptID下对应多个StageID
    private List<String> getCompletedStageIdList(String appLink, String completedAttemptId) throws IOException {
        String url = sparkHistoryUrl  + "/" + appLink + "/" + completedAttemptId + "/stages";
        // 发送请求
        String resultJsonStr = HttpUtil.get(url);

        List<JSONObject> stageJsonArr = JSON.parseArray(resultJsonStr, JSONObject.class);

        ArrayList<String> completedStageIdList = new ArrayList<>();
        for (JSONObject stageJsonObj : stageJsonArr) {
            if("COMPLETE".equals(stageJsonObj.getString("status"))) {
                completedStageIdList.add(stageJsonObj.getString("stageId"));
            }
        }

        return completedStageIdList;
    }

    // TODO 7 提取completed:true 的attemptID  一个yarnID对应一个成功的attemptID
    private String getCompletedAttemptId(String appLink) throws IOException {
        String url = sparkHistoryUrl + "/" + appLink;
        // 发送请求 待补充
        String resultJsonStr = HttpUtil.get(url);

        JSONObject jsonObject = JSON.parseObject(resultJsonStr);
        JSONArray attemptsJsonArr = jsonObject.getJSONArray("attempts");

        for (int i = 0; i < attemptsJsonArr.size(); i++) {
            JSONObject attemptJsonObj = attemptsJsonArr.getJSONObject(i);
            if(attemptJsonObj.getBoolean("completed")) {
                return attemptJsonObj.getString("attemptId");
            }
        }

        return null;
    }
}
